package com.bw.day0713.gmall;

import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.api.common.typeinfo.Types;

public class DWS_gd12_gender_state {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 创建示例数据源表
        tEnv.executeSql("create table dwd_user_join( " +
                "user_id string, " +
                "product_title string, " +
                "behavior_type string, " +
                "behavior_time string " +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_user_join_topic", "user_join"));

        // 使用Flink SQL进行分数计算
        Table scoreTable = tEnv.sqlQuery("SELECT " +
                "user_id," +
                "CASE " +
                "    WHEN behavior_type LIKE '%下单%' THEN 1 " +
                "    WHEN behavior_type LIKE '%加购%' THEN 0.5 " +
                "    WHEN behavior_type LIKE '%收藏%' THEN 0.4 " +
                "    WHEN behavior_type LIKE '%浏览%' THEN 0.2 " +
                "    ELSE 0 " +
                "END AS score " +
                "FROM dwd_user_join");

        // 将Table转换为DataStream
        DataStream<Tuple2<String, Double>> scoreStream = tEnv.toDataStream(scoreTable)
                .map(row -> Tuple2.of(
                        row.getFieldAs("user_id").toString(),
                        ((Number) row.getFieldAs("score")).doubleValue()
                ))
                .returns(Types.TUPLE(Types.STRING, Types.DOUBLE)); //

        // 使用DataStream API管理状态
        DataStream<Tuple2<String, Double>> keyByStream = scoreStream
                .keyBy(tuple -> tuple.f0);


        SingleOutputStreamOperator<Tuple2<String, Double>> scoreState = keyByStream.flatMap(new RichFlatMapFunction<Tuple2<String, Double>, Tuple2<String, Double>>() {
            private transient ValueState<Double> scoreState;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 初始化状态
                ValueStateDescriptor<Double> descriptor = new ValueStateDescriptor<>(
                        "scoreState", // 状态名称
                        Double.class  // 状态类型
                );
                scoreState = getRuntimeContext().getState(descriptor);
            }

            @Override
            public void flatMap(Tuple2<String, Double> input, Collector<Tuple2<String, Double>> out) throws Exception {

                // 获取当前状态值
                Double currentScore = scoreState.value();
                if (currentScore == null) {
                    currentScore = 0.0;
                }

                // 更新状态值
                currentScore += input.f1;
                scoreState.update(currentScore);

                // 输出结果
                out.collect(Tuple2.of(input.f0, currentScore));
            }
        });

//         打印结果
        scoreState.map(tuple -> Tuple2.of(
                tuple.f0,
                (double) Math.round(tuple.f1 * 100) / 100  // 保留两位小数
        )).print();
        // 执行作业
        env.execute("Flink SQL State Example");
    }

}
